package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/2/15 08:29
 */
public class Dwd_06_DwdTradeOrderRefund extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_06_DwdTradeOrderRefund().init(
            3007,
            2,
            "Dwd_06_DwdTradeOrderRefund"
        );
        
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1. 读取 ods_db
        readOdsDb(tEnv, "Dwd_06_DwdTradeOrderRefund");
        // 2. 读取字典表
        readBaseDic(tEnv);
        // 3. 过滤出退单表数据  insert
        Table orderRefundInfo = tEnv.sqlQuery("select " +
                                                  "data['id'] id,  " +
                                                  "data['user_id'] user_id,  " +
                                                  "data['order_id'] order_id,  " +
                                                  "data['sku_id'] sku_id,  " +
                                                  "data['refund_type'] refund_type,  " +
                                                  "data['refund_num'] refund_num,  " +
                                                  "data['refund_amount'] refund_amount,  " +
                                                  "data['refund_reason_type'] refund_reason_type,  " +
                                                  "data['refund_reason_txt'] refund_reason_txt,  " +
                                                  "data['create_time'] create_time,  " +
                                                  "pt,  " +
                                                  "ts  " +
                                                  "from ods_db " +
                                                  "where `database`='gmall2022' " +
                                                  "and `table`='order_refund_info' " +
                                                  "and `type`='insert' ");
        tEnv.createTemporaryView("order_refund_info", orderRefundInfo);
        
        // 4. 从 order_info 过滤退单数据
        Table orderInfo = tEnv.sqlQuery("select " +
                                            "data['id'] id,  " +
                                            "data['province_id'] province_id,  " +
                                            "`old`  " +
                                            "from ods_db " +
                                            "where `database`='gmall2022' " +
                                            "and `table`='order_info' " +
                                            "and `type`='update' " +
                                            "and `old`['order_status']='1002'" +
                                            "and `data`['order_status']='1005'");
        tEnv.createTemporaryView("order_info", orderInfo);
        // 5. 3 张表 join(4 张)
        Table result = tEnv.sqlQuery("select " +
                                         "ori.id,  " +
                                         "ori.user_id,  " +
                                         "ori.order_id,  " +
                                         "ori.sku_id,  " +
                                         "oi.province_id,  " +
                                         "date_format(ori.create_time,'yyyy-MM-dd') date_id,  " +
                                         "ori.create_time,  " +
                                         "ori.refund_type,  " +
                                         "dic1.dic_name refund_type_name,  " +
                                         "ori.refund_reason_type,  " +
                                         "dic2.dic_name refund_reason_type_name,  " +
                                         "ori.refund_reason_txt,  " +
                                         "ori.refund_num,  " +
                                         "ori.refund_amount,  " +
                                         "ori.ts  " +
                                         "from order_refund_info ori " +
                                         "join order_info oi " +
                                         "on ori.order_id=oi.id " +
                                         "join base_dic for system_time as of ori.pt dic1 " +
                                         "on ori.refund_type=dic1.dic_code " +
                                         "join base_dic for system_time as of ori.pt dic2 " +
                                         "on ori.refund_reason_type=dic2.dic_code ");
        
        // 5. 写出到 kafka 中
        tEnv.executeSql("create table dwd_trade_order_refund(  " +
                            "id string,  " +
                            "user_id string,  " +
                            "order_id string,  " +
                            "sku_id string,  " +
                            "province_id string,  " +
                            "date_id string,  " +
                            "create_time string,  " +
                            "refund_type_code string,  " +
                            "refund_type_name string,  " +
                            "refund_reason_type_code string,  " +
                            "refund_reason_type_name string,  " +
                            "refund_reason_txt string,  " +
                            "refund_num string,  " +
                            "refund_amount string,  " +
                            "ts bigint  " +
                            ")" + SQLUtil.getDDLKafkaSink(Constant.TOPIC_DWD_TRADE_ORDER_REFUND));
    
//        result.executeInsert("dwd_trade_order_refund");
        tEnv.executeSql("insert into dwd_trade_order_refund select * from " + result);
    }
}
/*
退单事务事实表

粒度: 商品粒度

order_refund_info:   insert
    join
order_info: update  order_status 1002->1005
    look up
base_dic  两次

ttl: 5s
 */